package com.ruby.bigtable.spark;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

@SuppressWarnings({ "unused", "unchecked", "resource", "serial" })
public class ClassClassic {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf()
				.setAppName("class classic")
				.setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		List<Tuple2<String, Integer>> scoreList = Arrays.asList(
				new Tuple2<String, Integer>("class1", 97),
				new Tuple2<String, Integer>("class2", 25),
				new Tuple2<String, Integer>("class3", 66),
				new Tuple2<String, Integer>("class2", 87),
				new Tuple2<String, Integer>("class2", 85),
				new Tuple2<String, Integer>("class1", 78),
				new Tuple2<String, Integer>("class3", 69)
				);
		
		JavaPairRDD<String, Integer> scoreRdd = sc.parallelizePairs(scoreList, 1);
		JavaPairRDD<String, Integer> resultRdd = scoreRdd.mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {

			public Tuple2<Integer, String> call(Tuple2<String, Integer> t) throws Exception {
				// TODO Auto-generated method stub
				return new Tuple2<Integer, String>(t._2, t._1);
			}
		}).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String,Integer>() {

			public Tuple2<String,Integer> call(Tuple2<Integer, String> t) throws Exception {
				// TODO Auto-generated method stub
				return new Tuple2<String,Integer>(t._2, t._1);
			}
		});
		resultRdd.foreach(new VoidFunction<Tuple2<String,Integer>>() {

			public void call(Tuple2<String, Integer> t) throws Exception {
				// TODO Auto-generated method stub
				System.out.println("class:"+t._1+" ,score:"+t._2);
			}
		});
		/**
		JavaPairRDD<String, Integer> scoreClassRdd = scoreRdd.reduceByKey(new Function2<Integer, Integer, Integer>(){

			public Integer call(Integer v1, Integer v2) throws Exception {
				// TODO Auto-generated method stub
				return v1+v2;
			}
			
		}).sortByKey();
		scoreClassRdd.foreach(new VoidFunction<Tuple2<String,Integer>>() {

			public void call(Tuple2<String, Integer> t) throws Exception {
				// TODO Auto-generated method stub
				System.out.println("class:"+t._1+" ,score:"+t._2);
			}
		});
		/**
		JavaPairRDD<String, Iterable<Integer>> scoreGroupRdd = scoreRdd.groupByKey();
		scoreGroupRdd.foreach(new VoidFunction<Tuple2<String,Iterable<Integer>>>() {
			
			public void call(Tuple2<String, Iterable<Integer>> t) throws Exception {
				// TODO Auto-generated method stub
				String classid = t._1;
				System.out.println("class:"+classid);
				Iterator<Integer> scoreList = t._2.iterator();
				while (scoreList.hasNext()) {
					Integer score = scoreList.next();
					System.out.println("score:"+score);
				}
				System.out.println("------------------------------");
			}
		});
		**/
	}

}
